use crate::pipeline::MutExecutor;

use super::*;
use std::{fs::OpenOptions, io::Read, time::Instant};
use serde::{Deserialize, Serialize};
use walkdir::WalkDir;

pub const READ_STAGE_NAME:&str = "read";

#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct ReaderConfig {
    continous_split:bool,
    split_min_len:usize,
    read_buffer_size:usize,
    split_buffer_size:usize,
}

impl ReaderConfig {
    pub fn from_source(source:&SourcePool) -> Option<Self> {
        let config = source.get_config();
        let read_buffer_size = config.get("read_buffer_size").ok()?;
        let split_buffer_size = config.get("split_buffer_size").ok()?;
        let continous_split= config.get("continuous_split").ok()?;
        let split_min_len = source.chunking.min_len();
        Some(Self { 
            continous_split, 
            split_min_len,
            read_buffer_size, 
            split_buffer_size 
        })
    }

    pub fn new() -> Self {
        Self { 
            continous_split:false,
            split_min_len:0x1000,
            read_buffer_size:0x1000, 
            split_buffer_size:0x100_000  
        }
    }

    pub fn build(self) -> PathReader {
        PathReader {
            continuous_split:self.continous_split,
            split_min_len:self.split_min_len,
            read_buffer_size: self.read_buffer_size, 
            split_buffer_size: self.split_buffer_size,
            current_file_id: 0,
            read_time:0,
            read_size:0,
        }
    }
}

pub struct PathReader {
    continuous_split:bool,
    split_min_len:usize,
    read_buffer_size:usize,
    split_buffer_size:usize,
    current_file_id:usize,
    read_time:u128,
    read_size:usize,
}

impl PathReader {
    pub fn get_file_id(&mut self) -> usize {
        let alloc_id = self.current_file_id;
        self.current_file_id += 1;
        alloc_id
    }

    pub fn read_path<F:FnMut(FileBlock)>(&mut self, path:String, mut produce_file_block:F) -> BackupResult<()> {
        for entry in WalkDir::new(path) {
            let file_name = entry?.path().to_path_buf();
            let metadata = std::fs::metadata(&file_name)?;
            if metadata.is_dir() { continue; }
            let file_id = self.get_file_id();
            let file_name_string = file_name.to_str().unwrap().to_string();
            let file_name_kind = FileKind::Name(file_name_string);
            produce_file_block(file_name_kind.with_id(file_id));
            let mut file_description = OpenOptions::new()
                .read(true)
                .open(file_name)?;
            let mut byte_buffer_id = 0;
            let mut file_unread_size = metadata.len() as usize;
            let mut first_read_flag = true;
            while file_unread_size != 0 {
                let byte_buffer_size = std::cmp::min(file_unread_size, self.split_buffer_size);
                let extra_buffer_size = if first_read_flag || !self.continuous_split {
                    first_read_flag = false;
                    0
                } else {
                    self.split_min_len
                };
                let mut read_buffer = BytesMut::with_capacity(byte_buffer_size + extra_buffer_size);
                unsafe { read_buffer.set_len(byte_buffer_size + extra_buffer_size); }
                if extra_buffer_size != 0 {
                    let tmp_buf = read_buffer.split_to(extra_buffer_size);
                    let tmp_buf_kind = FileKind::Tmp(byte_buffer_id, tmp_buf);
                    produce_file_block(tmp_buf_kind.with_id(file_id));
                }
                while read_buffer.len() != 0 {
                    let read_buffer_size = std::cmp::min(read_buffer.len(), self.read_buffer_size);
                    let start_time = Instant::now();
                    file_description.read_exact(&mut read_buffer[..read_buffer_size])?;
                    self.read_time += start_time.elapsed().as_micros();
                    self.read_size += read_buffer_size;
                    let ok_buffer = read_buffer.split_to(read_buffer_size);
                    let file_data_kind = FileKind::Data(byte_buffer_id, ok_buffer);
                    produce_file_block(file_data_kind.with_id(file_id));
                }
                file_unread_size -= byte_buffer_size;
                byte_buffer_id += 1;
            }
            produce_file_block(FileKind::FileEnd.with_id(file_id));
        }
        Ok(())
    }

    pub fn get_read_size(&self) -> usize {
        self.read_size
    }

    pub fn get_read_time(&self) -> u128 {
        self.read_time
    }
}

impl Evaluated for PathReader {
    fn register(&self, evaluation:&Evaluation) {
        let mut eva = evaluation.lock();
        eva.register(BackupEvaluation::Space,"data size(B)");
        eva.register(BackupEvaluation::Time,"read time(us)");
    }

    fn set_value(&self, evaluation:&Evaluation) {
        let mut eva = evaluation.lock();
        eva.set("data size(B)", self.get_read_size() as u64);
        eva.set("read time(us)", self.get_read_time());
    }
}

impl MutExecutor<String, FileBlock> for PathReader {
    type Error = BackupError;
    fn exec<'scope, F:FnMut(FileBlock)>(&'scope mut self, input:String, output:F) -> Result<(), Self::Error> {
        self.read_path(input, output)
    }
}